package cn.demoncat.util.kafka.util;

import cn.demoncat.util.kafka.entity.KafkaMqCallback;
import cn.demoncat.util.lang.JsonUtil;
import cn.demoncat.util.lang.StringUtil;
import cn.demoncat.util.lang.constant.StringConstant;
import cn.demoncat.util.mq.entity.MqCallback;
import cn.demoncat.util.mq.entity.MqDelayMessage;
import cn.demoncat.util.mq.entity.MqMessage;
import cn.demoncat.util.mq.util.MqUtil;
import cn.demoncat.util.redis.entity.RedisDelay;
import cn.demoncat.util.redis.util.RedisDelayUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * Kafka 工具类
 *
 * 1、发送消息到MQ
 * 2、通过Redis ZSet 缓存延时消息
 *
 * @author 延晓磊
 *
 * @since 2018年8月10日
 *
 * @see <a href="https://kafka.apache.org">Kafka</a>
 * @see <a href="https://kafka.apachecn.org">Kafka 中文</a>
 * @see <a href="https://docs.spring.io/spring-kafka/docs/current/reference/html/">Spring Kafka</a>
 */
@Component("mqUtil")
public class KafkaMqUtil extends MqUtil {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    @Resource
    private RedisDelayUtil redisDelayUtil;

    // 延时队列：将所有延时消息暂存在Redis ZSet中，然后定时监控并补发
    private static final RedisDelay DELAY = new RedisDelay("MQ");

    private static final Logger LOG = LoggerFactory.getLogger(KafkaMqUtil.class);

    /**
     * 发送消息
     *
     * @param topic	    主题
     * @param key	    Key，均衡partition
     * @param msg	    消息（MqMessage）
     * @param callback	成功/失败回调
     * @return
     *
     * @author 延晓磊
     *
     * @since 2018年8月10日
     */
    @Override
    protected void send(String topic, String key, MqMessage msg, MqCallback callback) {
        if (StringUtil.isBlank(key)) {
            kafkaTemplate.send(topic, JsonUtil.toStr(msg)).addCallback(new KafkaMqCallback(callback));
        }else {
            kafkaTemplate.send(topic, key, JsonUtil.toStr(msg)).addCallback(new KafkaMqCallback(callback));
        }
    }

    /**
     * 发送延时消息
     *
     * 1、将消息暂存在Redis ZSet中
     * 2、通过执行monitorDelayMessage()监控和补发
     *
     * @param message
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    @Override
    protected void sendDelay(MqDelayMessage message) {
        if (message.getSendTime() != null) {
            redisDelayUtil.add(DELAY.getKey(), JsonUtil.toString(message), message.getSendTime());
        }else{
            redisDelayUtil.add(DELAY.getKey(), JsonUtil.toString(message), message.getDelaySeconds());
        }
    }

    /**
     * 监控延时消息（配置调用；全局默认队列，建议由“任务调度系统”执行）
     *
     * 1、间隔扫描Redis ZSet
     * 2、将到期消息取出并重新发送，发送成功时删除
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    @Override
    public void monitorDelayMessage() {
        redisDelayUtil.consume(DELAY, msg ->{
            // 补发延时消息
            MqDelayMessage message = JsonUtil.parse(msg, MqDelayMessage.class);
            this.send(message.getTopic(), message.getKey(), message.getMessage(), LOG);
        });
    }

    /**
     * 监控延时消息（配置调用；自定义队列）
     *
     * 1、间隔扫描Redis ZSet
     * 2、将到期消息取出并重新发送，发送成功时删除
     *
     * @param name	队列名称
     * @param seconds 消费间隔
     *
     * @author 延晓磊
     * @since 2021年02月23日
     */
    @Override
    public void monitorDelayMessage(String name, int seconds) {
        redisDelayUtil.consume(new RedisDelay(DELAY.getKey() + StringConstant.COLON + name, seconds), msg ->{
            // 补发延时消息
            MqDelayMessage message = JsonUtil.parse(msg, MqDelayMessage.class);
            this.send(message.getTopic(), message.getKey(), message.getMessage(), LOG);
        });
    }
}
